package org.raymond.iworks.study.mq.rabbitmq.stream.official;

import com.rabbitmq.stream.*;

import java.util.UUID;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicLong;
import java.util.stream.IntStream;

/**
 * @author raymond
 * @version V1.0
 * @Description: TODO
 */

/**
 * https://blog.rabbitmq.com/posts/2021/07/connecting-to-streams/
 * 流的拓扑结构:
 *   流是可复制和持久的，它由leader Erlang进程和replica Erlang进程组成。这些进程分布在RabbitMQ集群的节点上
 *   流是一种复制和持久的数据结构。它有一个leader进程，它接受写操作和副本，它可以向应用程序发送消息
 *   只有leader进程处理写操作，如入站消息。流的任何成员(leader和副本)都可以向客户端应用程序分派消息。
 *   使用stream protocol发布到stream的应用程序可以连接到集群中的任何节点:消息将自动从连接节点转到承载leader进程的节点。
 * 如果连接和stream leader不在同一节点上，操作就不是最佳的。发布到stream的应用程序应该连接到承载stream leader的节点，
 * 以避免额外的network hop。
 *   对于消费者，情况略有不同。使用stream protocol，使用sendfile将消息分派给应用程序:包含消息的文件块直接从节点文件系统
 * 发送到网络socket，而不通过用户空间。此优化要求消费者连接到的节点承载stream的一个成员。无论这个成员是leader还是replica
 * 都无关紧要，只要数据在文件系统上，就可以通过network socket使用sendfile。
 *   对于消费者来说，这个约束似乎并不太苛刻:在上面的图表中，每个节点都有一个流成员，因此应用程序可以连接到任何节点进行消费。
 * 但是想象一下，一个5节点集群的流使用2的复制因子:每个流将只在5个节点中的3个节点上有成员。在这种情况下，消费应用程序必须
 * 适当地选择它们的节点。
 *
 * 彬彬有礼的客户:
 *     因此，发布者可以连接到集群的任何节点，它将始终到达leader进程。消费者必须连接到承载目标流成员的节点，该成员可以是leader或replica。
 * 客户端应用程序可以遵循这些规则，但应尽可能执行以下最佳实践
 * 发布者应该始终连接到承载目标流的leader的节点
 * 消费者应该始终连接到承载目标流的replica的节点
 *     如前所述，直接连接到流leader的节点避免了网络跳，因为消息无论如何都必须进入领导进程。总是使用replica进行消费呢?
 * 它只是卸载了leader进程，它(leader)已经在忙于处理所有的写操作了.
 *
 * 跨集群的流:
 *     在继续讨论流协议的元数据命令之前，让我们先澄清一下流(复数)是如何在RabbitMQ集群的节点间传播的。我们提到，
 * 流有一个位于节点上的leader Erlang进程和位于其他节点上的replica Erlang进程。对于几个流，所有这些Erlang进程
 * (对于Erlang进程来说是“非常轻量级的线程”)分布在集群节点上，并且在任何情况下，一个给定的RabbitMQ节点都不应该托管
 * 所有的流leader。
 * 流leader分布在集群的节点上。这意味着给定的节点不必在某一点上包含所有的leader。
 *     leader在集群中分布的方式取决于创建流时的“leader定位策略”。默认策略是“最少领导者”，这意味着选择具有最少数量的
 * 流leader的节点作为新流的领导者。还有其他一些策略，但这已经超出了本文的讨论范围
 *
 * 元数据命令:
 *     一旦客户端应用程序了解了流的拓扑结构，它就可以连接到适当的节点来使用它:承载要发布的流leader的节点和承载要消费
 * 的流replica的节点
 *
 * Advertiesd_host:
 *     advertised_host和advertised_port设置应该有助于解决客户端应用程序由于不正确的流元数据而无法连接到节点的所有问题。
 * 如果你部署了一个带有容器化节点的RabbitMQ集群，并且使用了流，一定要记住它们。
 *     在一个常见的用例中，这种发现机制可能会出现问题:当负载平衡器位于客户机应用程序和集群节点之间时。
 *
 *     客户机应用程序仍然可以始终连接到负载均衡器，并最终连接到适当的节点。它是如何做到的呢?两个步骤:
 * 使用元数据命令，但忽略该信息，始终连接到负载均衡器
 * 重试连接，直到它连接到适当的节点
 *     您可能想知道，一旦建立了连接，如何查找节点的信息?节点的“坐标”(主机名和端口，或者如果配置了，则为advertised_host
 * 和advertised_port)在流协议连接中可用。因此客户机应用程序可以知道它连接到哪个节点。
 *     这意味着在使用负载平衡器时不应该配置advertised_host和advertised_port。在这种情况下，元数据命令返回的节点的
 * “坐标”不用于连接，因为客户端总是连接到负载均衡器。它们用于将负载均衡器提供的连接与客户机期望的节点相关联，主机名
 * 非常适合于此。
 *
 * 使用带有负载平衡器的流Java客户端
 *     流Java客户机提供了一个AddressResolver扩展点。每当创建新连接时都会使用它:从传入的Address(基于元数据查询要连接
 * 到的节点)，地址解析器可以提供一些逻辑来计算要使用的实际地址。默认实现只返回给定的地址。如果您想在使用负载均衡器时实现
 * 上述解决方案，请始终返回负载均衡器的地址
 * Address entryPoint = new Address("my-load-balancer", 5552);
 * Environment environment = Environment.builder()
 *     .host(entryPoint.host())
 *     .port(entryPoint.port())
 *     .addressResolver(address -> entryPoint)
 *     .build();
 */
public class SampleApplication {
    public static void main(String[] args) throws Exception {
        // tag::sample-environment[]
        System.out.println("Connecting...");
        Environment environment = Environment.builder()
                .host("192.168.0.125")
                .port(5552)
                .username("cloud")
                .password("cloud123")
                .virtualHost("/cloud")
                //.addressResolver(address -> new Address("192.168.0.125", 5552))
                .build();  // <1>
        String stream = UUID.randomUUID().toString();
        environment.streamCreator().stream(stream).create();  // <2>
        // end::sample-environment[]
        // tag::sample-publisher[]
        System.out.println("Starting publishing...");
        int messageCount = 10000;
        CountDownLatch publishConfirmLatch = new CountDownLatch(messageCount);
        Producer producer = environment.producerBuilder()  // <1>
                .stream(stream)
                .build();
        IntStream.range(0, messageCount)
                .forEach(i -> producer.send(  // <2>
                        producer.messageBuilder()                    // <3>
                                .addData(String.valueOf(i).getBytes())   // <3>
                                .build(),                                // <3>
                        confirmationStatus -> publishConfirmLatch.countDown()  // <4>
                ));
        publishConfirmLatch.await(10, TimeUnit.SECONDS);  // <5>
        producer.close();  // <6>
        System.out.printf("Published %,d messages%n", messageCount);
        // end::sample-publisher[]

        // tag::sample-consumer[]
        System.out.println("Starting consuming...");
        AtomicLong sum = new AtomicLong(0);
        CountDownLatch consumeLatch = new CountDownLatch(messageCount);
        Consumer consumer = environment.consumerBuilder()  // <1>
                .stream(stream)
                .offset(OffsetSpecification.first()) // <2>
                .messageHandler((offset, message) -> {  // <3>
                    sum.addAndGet(Long.parseLong(new String(message.getBodyAsBinary())));  // <4>
                    consumeLatch.countDown();  // <5>
                })
                .build();

        consumeLatch.await(10, TimeUnit.SECONDS);  // <6>

        System.out.println("Sum: " + sum.get());  // <7>

        consumer.close();  // <8>
        // end::sample-consumer[]

        // tag::sample-environment-close[]
        environment.deleteStream(stream);  // <1>
        environment.close();  // <2>
        // end::sample-environment-close[]
    }
}
